package com.whz.mlc.hotkey.impl;

import com.google.common.collect.Maps;
import com.whz.generic.utils.IpUtils;
import com.whz.mlc.cache.data.MlcCacheEntityWrapper;
import com.whz.mlc.cache.second.redis.client.MlcRedisClient;
import com.whz.mlc.hotkey.HotKeyCounter;
import com.whz.mlc.hotkey.HotKeyProperty;
import lombok.Builder;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.BoundValueOperations;
import org.springframework.data.redis.core.BoundZSetOperations;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.util.Assert;

import java.time.Duration;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * 基于redis的滑动时间窗口计数器
 *
 * @author whz
 * 2021/8/4
 */
@Slf4j
@Builder
public class SlidingWindowCounter implements HotKeyCounter {
    public static final int DURATION_OF_HOLD_LOCK_SECONDS = 60;
    // 滑动窗口时间间隔s
    private HotKeyProperty config;

    private String cacheName;

    @Override
    public void hotKeyCount(String hotKey, Object value) {
        this.hotKeyCountSize(hotKey, value, 1);
    }

    @Override
    public void hotKeyCountSize(String hotKey, Object value, int size) {
        // 统计hotKey热度
        Long count = countHot(hotKey, size);
        // hotKey当前热度
        MlcCacheEntityWrapper<Object> wrapper = new MlcCacheEntityWrapper<>();
        wrapper.setKey(hotKey);
        wrapper.setValue(value);
        // 热度排序
        hotKeySort(wrapper, count);
    }

    @SneakyThrows
    @Override
    public Map<String, Object> currentHotKeys() {
        Map<String, Object> maps = Maps.newHashMap();
        if (tryLock()) {
            int hotKeyQueueSize = config.getHotKeyQueueSize() < 1? 100 : config.getHotKeyQueueSize();
            try (Cursor<ZSetOperations.TypedTuple<Object>> cursor = getKeysSet().scan(
                    ScanOptions.scanOptions().count(hotKeyQueueSize).build())) {
                while (cursor.hasNext()) {
                    ZSetOperations.TypedTuple<Object> v = cursor.next();
                    if (v.getValue() instanceof MlcCacheEntityWrapper) {
                        MlcCacheEntityWrapper<?> value = (MlcCacheEntityWrapper<?>) v.getValue();
                        maps.put(value.getKey(), value.getValue());
                    }
                }
                log.info("获取当前hot keys count:{}", maps.size());
            } finally {
                unlock();
            }
        }
        return maps;
    }

    @Override
    public Boolean delete(String key) {
        return Boolean.TRUE.equals(getTemplate().boundZSetOps(getCKey(key)).getOperations().delete(key)) &&
               Boolean.TRUE.equals(getKeysSet().getOperations().delete(key));
    }

    private RedisTemplate<String, Object> getTemplate() {
        return MlcRedisClient.getDefaultClient();
    }


    private void hotKeySort(MlcCacheEntityWrapper<?> wrapper, Long count) {
        Assert.isTrue(config.getHotKeyQueueSize() > 0, "hotKeyQueueSize must greater than 0");
        long currentMs = System.currentTimeMillis();
        long maxScoreMs = currentMs - config.getWindowPeriodSeconds() * 1000L;

        try {
            // 单独的有序set
            String k = getHKey();
            getTemplate().watch(k);
            getTemplate().multi();
            BoundZSetOperations<String, Object> zSetOperations = getTemplate().boundZSetOps(k);
            // 清除超过时间窗周期的成员
            zSetOperations.removeRangeByScore(0, maxScoreMs);
            zSetOperations.add(wrapper, count);
            // 保留指定数量的Key
            zSetOperations.removeRange(config.getHotKeyQueueSize(), Integer.MAX_VALUE);
            zSetOperations.expire(config.getWindowPeriodSeconds(), TimeUnit.SECONDS);
            getTemplate().exec();
        } catch (Exception exception) {
            log.error("hotkey排序异常.", exception);
            getTemplate().discard();
        }
    }

    private Long countHot(String hotKey, int count) {
        try {
            String k = getCKey(hotKey);
            BoundValueOperations<String, Object> zSetOperations = getTemplate().boundValueOps(k);
            // 热点key计数
            Long expire = zSetOperations.getExpire();
            if (expire == null || expire <= 0) {
                zSetOperations.expire(config.getWindowPeriodSeconds(), TimeUnit.SECONDS);
            }
            return zSetOperations.increment(count);
        } catch (Exception exception) {
            log.error("滑动窗口hotkey计数异常.", exception);
        }
        return null;
    }

    private String getCKey(String key) {
        return BASE_SPACE + cacheName + ":windows:" + key;
    }

    private String getHKey() {
        return BASE_SPACE + cacheName + ":hot-keys-zset";
    }

    private String getLock() {
        return BASE_SPACE + cacheName + ":lock";
    }

    private BoundZSetOperations<String, Object> getKeysSet() {
        return getTemplate().boundZSetOps(getHKey());
    }

    // 简单的lock
    private boolean tryLock() {
        return Boolean.TRUE.equals(getTemplate().opsForValue().setIfAbsent(getLock(), IpUtils.getLocalIpNum(),
                                                                           Duration.ofSeconds(
                                                                                   DURATION_OF_HOLD_LOCK_SECONDS)));
    }

    private void unlock() {
        long localIp = IpUtils.getLocalIpNum();
        Object obj = getTemplate().opsForValue().get(getLock());

        if (!Objects.isNull(obj) && obj.equals(String.valueOf(localIp))) {
            getTemplate().delete(getLock());
        }
    }

}